package sparkSql

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.{SparkConf, SparkContext}

object Demo1 {
  case class Person(id:Int, name:String, gender:String, age:Int)
  def main(args: Array[String]): Unit = {
    //创建SparkConf()并设置App名称
    val conf = new SparkConf().setMaster("local[2]").setAppName("Spark SQL")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")
    val spark = SparkSession
      .builder()
      .config(conf)
      .getOrCreate()

    import spark.implicits._

//    val df: DataFrame = spark.read.json("data/employee.json")
    val df: DataFrame = spark.read.json("file:///opt/data/employee.json")
    df.createOrReplaceTempView("persons")
    spark.udf.register("addName", (x:String)=> "user"+x)

    println("----------------查询 所有 数据---------------------")
    df.show()
    println("----------------查询 所有 数据， 并 去除 重复 的 数据---------------------")
    df.distinct().show()
    println("----------------查询 所有 数据， 打印 时 去除 id 字段---------------------")
    spark.sql("select name,age from persons").show()
    println("----------------筛选 出 age> 30 的 记录---------------------")
    spark.sql("select * from persons where age > 30").show()
    println("----------------将 数据 按 age 分组---------------------")
    df.groupBy("age").count().show()
    println("----------------将 数据 按 name 升序 排列---------------------")
    df.sort("name").show()
    println("----------------取出 前 3 行 数据---------------------")
    df.limit(3).show()
    println("----------查询 所有 记录 的 name 列， 并为 其 取 别名 为 username-----------")
    df.select("name").toDF("username").show()
    println("----------查询 年龄 age 的 平均值-----------")
    spark.sql("select avg(age) from persons").show()
    println("----------查询 年龄 age 的 最小值-----------")
    spark.sql("select min(age) from persons").show()

    /**
      * 编程 实 现将 RDD 转换 为 DataFrame 源 文件 内容 如下（ 包含 id, name, age）：
      * 1, Ella, 36
      * 2, Bob, 29
      * 3, Jack, 29
      * 请将 数据 复制 保存 到 Linux 系统 中， 命名为 employee. txt， 实现 从 RDD 转换 得
      * 到 DataFrame， 并按“ id: 1, name: Ella, age: 36” 的 格式 打 印出 DataFrame 的 所有
      * 数据。 请 写出 程序 代码。
      */

//    val path = "data/employee.txt"
    val path = "file:///opt/data/employee.txt"
    val df2: DataFrame = spark.read.textFile(path).map {
      x =>val para = x.split(",");
        (para(0).trim.toInt, para(1), para(2).trim.toInt)
    }.toDF("id", "name", "age")
    df2.createOrReplaceTempView("employees")
    spark.udf.register("addId", (x:Int)=> "id:"+x)
    spark.udf.register("addName", (x:String)=> "name:"+x)
    spark.udf.register("addAge", (x:Int)=> "age:"+x)
    println("----------按“ id: 1, name: Ella, age: 36” 的 格式 打 印出 DataFrame 的 所有数据-----------")
    spark.sql("select addId(id),addName(name),addAge(age) from employees").show()

    /**
      * 3.编程 实现 利用 DataFrame 读写 MySQL 的 数据
      */

    val rdd: RDD[(Int, String, String, Int)] =
      sc.makeRDD(List((3,"Mary","F",26),(4,"Tom","M",23)))
    val df3= rdd.map(x=>Person(x._1,x._2,x._3,x._4)).toDF()
    df3.write
      .format("jdbc")
      .mode("append")
      .option("url", "jdbc:mysql://flum:3306/sparktest")
      .option("dbtable", "employee")
      .option("user", "root")
      .option("password", "123456")
      .save()
    val jdbcDF = spark.read
      .format("jdbc")
      .option("url", "jdbc:mysql://flum:3306/sparktest")
      .option("dbtable", "employee")
      .option("user", "root")
      .option("password", "123456")
      .load()
    jdbcDF.createOrReplaceTempView("peoples")
    println("--------age 的 最大值 和 age 的 总和------------")
    spark.sql("select max(age) from peoples").show()
    spark.sql("select sum(age) from peoples").show()

  }
}
